package com.heima.article.stream;

import com.heima.article.config.KafkaStreamListener;
import com.heima.common.constants.MQConstants;
import com.heima.model.article.dtos.ArticleVisitStreamMsg;
import com.heima.model.search.dtos.UpdateArticleMsg;
import com.heima.utils.common.JsonUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;

import java.util.Arrays;


/**
 * 热点实时刷新流式处理程序
 */
@Component
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {
    @Override
    public String listenerTopic() {
        return MQConstants.HOT_ARTICLE_OUTPUT_TOPIC;
    }

    @Override
    public String sendTopic() {
        return MQConstants.HOT_ARTICLE_INPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        /**
         * 原始数据：
         *    key       value
         *             {"articleId":1001,"type":LIKES}    200次
         *             {"articleId":1002,"type":LIKES}    100次
         *             {"articleId":1001,"type":COMMENT}   50
         */

        /**
         * 目标数据：
         *    key       value
         *             {"articleId":1001","like": 200}   1次
         *             {"articleId":1002","like": 100}   1次
         *             {"articleId":1001","comment":50}  1次
         *
         */

        /**
         * 第一步：格式处理   要求格式：1001:LIKES
         * 第二步：将value赋值给key
         * 第三步：用key分组
         *      key                 value
         *     1001:LIKES
         *     1002:LIKES
         *     1001:COMMENT
         * 第四步：统计结果
         *      key                 value
         *     1001:LIKES            200
         *     1002:LIKES            100
         *     1001:COMMENT          50
         */

        KTable<Object, Long> kTable = stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) {
                UpdateArticleMsg msg = JsonUtils.toBean(value, UpdateArticleMsg.class);
                String msgStr = msg.getArticleId() + ":" + msg.getType().name();
                return Arrays.asList(msgStr);
            }
        }).map(new KeyValueMapper<String, Object, KeyValue<?, ?>>() {

            @Override
            public KeyValue<?, ?> apply(String key, Object value) {
                return new KeyValue<>(value, value);
            }
        }).groupByKey().count(Materialized.as("total"));

        KStream kStream = kTable.toStream().map((key, value) -> {
            ArticleVisitStreamMsg streamMsg = new ArticleVisitStreamMsg();
            String[] array = key.toString().split(":");
            streamMsg.setArticleId(Long.valueOf(array[0]));
            switch (UpdateArticleMsg.UpdateArticleType.valueOf(array[1])) {
                case VIEWS:
                    streamMsg.setView(value);
                    break;
                case COLLECTION:
                    streamMsg.setCollect(value);
                    break;
                case COMMENT:
                    streamMsg.setComment(value);
                    break;
                case LIKES:
                    streamMsg.setLike(value);

            }
            return new KeyValue<>(key, JsonUtils.toString(streamMsg));
        });

        return kStream;

    }

}
